{
 "cells": [
  {
   "cell_type": "markdown",
   "source": [
    "在日常工作中，我们常常会用到需要周期性执行的任务，一种方式是采用 Linux 系统自带的 crond 结合命令行实现，另外一种方式是直接使用Python。\n",
    "\n",
    "最近我整理了一下 Python 定时任务的实现方式，内容较长，建议收藏后学习，梳理不易，有所收获，点赞支持。\n",
    "\n",
    "我们开始学习吧！\n",
    "\n",
    "# 目录\n",
    "- 利用while True: + sleep()实现定时任务\n",
    "- 使用Timeloop库运行定时任务\n",
    "- 利用threading.Timer实现定时任务\n",
    "- 利用内置模块sched实现定时任务\n",
    "- 利用调度模块schedule实现定时任务\n",
    "- 利用任务框架APScheduler实现定时任务\n",
    "- 使用分布式消息系统Celery实现定时任务\n",
    "- 使用数据流工具Apache Airflow实现定时任务"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 1. 利用while True: + sleep()实现定时任务\n",
    "位于 time 模块中的 sleep(secs) 函数，可以实现令当前执行的线程暂停 secs 秒后再继续执行。所谓暂停，即令当前线程进入阻塞状态，当达到 sleep() 函数规定的时间后，再由阻塞状态转为就绪状态，等待 CPU 调度。\n",
    "基于这样的特性我们可以通过while死循环+sleep()的方式实现简单的定时任务。\n",
    "代码示例："
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "import datetime\n",
    "import time\n",
    "def time_printer():\n",
    "    now = datetime.datetime.now()\n",
    "    ts = now.strftime('%Y-%m-%d %H:%M:%S')\n",
    "    print('do func time :', ts)\n",
    "def loop_monitor():\n",
    "    while True:\n",
    "        time_printer()\n",
    "        time.sleep(5)  # 暂停5秒\n",
    "if __name__ == \"__main__\":\n",
    "    loop_monitor()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### 主要缺点：\n",
    "只能设定间隔，不能指定具体的时间，比如每天早上8:00\n",
    "sleep 是一个阻塞函数，也就是说 sleep 这一段时间，程序什么也不能操作。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 2. 使用Timeloop库运行定时任务\n",
    "Timeloop是一个库，可用于运行多周期任务。这是一个简单的库，它使用decorator模式在线程中运行标记函数。\n",
    "\n",
    "示例代码："
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "import time\n",
    "from timeloop import Timeloop\n",
    "from datetime import timedelta\n",
    "tl = Timeloop()\n",
    "@tl.job(interval=timedelta(seconds=2))\n",
    "def sample_job_every_2s():\n",
    "    print(\"2s job current time : {}\".format(time.ctime()))\n",
    "@tl.job(interval=timedelta(seconds=5))\n",
    "def sample_job_every_5s():\n",
    "    print(\"5s job current time : {}\".format(time.ctime()))\n",
    "@tl.job(interval=timedelta(seconds=10))\n",
    "def sample_job_every_10s():\n",
    "    print(\"10s job current time : {}\".format(time.ctime()))"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 3. 利用threading.Timer实现定时任务\n",
    "threading 模块中的 Timer 是一个非阻塞函数，比 sleep 稍好一点，timer最基本理解就是定时器，我们可以启动多个定时任务，这些定时器任务是异步执行，所以不存在等待顺序执行问题。\n",
    "\n",
    "Timer(interval, function, args=[ ], kwargs={ })\n",
    "interval: 指定的时间\n",
    "function: 要执行的方法\n",
    "args/kwargs: 方法的参数\n",
    "代码示例：\n",
    "\n",
    "备注：Timer只能执行一次，这里需要循环调用，否则只能执行一次"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 4. 利用内置模块sched实现定时任务\n",
    "sched模块实现了一个通用事件调度器，在调度器类使用一个延迟函数等待特定的时间，执行任务。同时支持多线程应用程序，在每个任务执行后会立刻调用延时函数，以确保其他线程也能执行。\n",
    "\n",
    "class sched.scheduler(timefunc, delayfunc)这个类定义了调度事件的通用接口，它需要外部传入两个参数，timefunc是一个没有参数的返回时间类型数字的函数(常用使用的如time模块里面的time)，delayfunc应该是一个需要一个参数来调用、与timefunc的输出兼容、并且作用为延迟多个时间单位的函数(常用的如time模块的sleep)。\n",
    "\n",
    "代码示例："
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "import datetime\n",
    "import time\n",
    "import sched\n",
    "def time_printer():\n",
    "    now = datetime.datetime.now()\n",
    "    ts = now.strftime('%Y-%m-%d %H:%M:%S')\n",
    "    print('do func time :', ts)\n",
    "    loop_monitor()\n",
    "def loop_monitor():\n",
    "    s = sched.scheduler(time.time, time.sleep)  # 生成调度器\n",
    "    s.enter(5, 1, time_printer, ())\n",
    "    s.run()\n",
    "if __name__ == \"__main__\":\n",
    "    loop_monitor()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "scheduler对象主要方法:\n",
    "enter(delay, priority, action, argument)，安排一个事件来延迟delay个时间单位。\n",
    "cancel(event)：从队列中删除事件。如果事件不是当前队列中的事件，则该方法将跑出一个ValueError。\n",
    "run()：运行所有预定的事件。这个函数将等待(使用传递给构造函数的delayfunc()函数)，然后执行事件，直到不再有预定的事件。\n",
    "个人点评：比threading.Timer更好，不需要循环调用。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 5. 利用调度模块schedule实现定时任务\n",
    "schedule是一个第三方轻量级的任务调度模块，可以按照秒，分，小时，日期或者自定义事件执行时间。schedule允许用户使用简单、人性化的语法以预定的时间间隔定期运行Python函数(或其它可调用函数)。\n",
    "\n",
    "先来看代码，是不是不看文档就能明白什么意思？"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "import schedule\n",
    "import time\n",
    "def job():\n",
    "    print(\"I'm working...\")\n",
    "schedule.every(10).seconds.do(job)\n",
    "schedule.every(10).minutes.do(job)\n",
    "schedule.every().hour.do(job)\n",
    "schedule.every().day.at(\"10:30\").do(job)\n",
    "schedule.every(5).to(10).minutes.do(job)\n",
    "schedule.every().monday.do(job)\n",
    "schedule.every().wednesday.at(\"13:15\").do(job)\n",
    "schedule.every().minute.at(\":17\").do(job)\n",
    "while True:\n",
    "    schedule.run_pending()\n",
    "    time.sleep(1)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 装饰器：通过 @repeat() 装饰静态方法\n",
    "\n",
    "import time\n",
    "from schedule import every, repeat, run_pending\n",
    "@repeat(every().second)\n",
    "def job():\n",
    "    print('working...')\n",
    "    \n",
    "while True:\n",
    "    run_pending()\n",
    "    time.sleep(1)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 传递参数：\n",
    "\n",
    "import schedule\n",
    "def greet(name):\n",
    "    print('Hello', name)\n",
    "schedule.every(2).seconds.do(greet, name='Alice')\n",
    "schedule.every(4).seconds.do(greet, name='Bob')\n",
    "while True:\n",
    "    schedule.run_pending()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 装饰器同样能传递参数：\n",
    "\n",
    "from schedule import every, repeat, run_pending\n",
    "@repeat(every().second, 'World')\n",
    "@repeat(every().minute, 'Mars')\n",
    "def hello(planet):\n",
    "    print('Hello', planet)\n",
    "while True:\n",
    "    run_pending()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 取消任务：\n",
    "\n",
    "import schedule\n",
    "i = 0\n",
    "def some_task():\n",
    "    global i\n",
    "    i += 1\n",
    "    print(i)\n",
    "    if i == 10:\n",
    "        schedule.cancel_job(job)\n",
    "        print('cancel job')\n",
    "        exit(0)\n",
    "job = schedule.every().second.do(some_task)\n",
    "while True:\n",
    "    schedule.run_pending()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 运行一次任务：\n",
    "\n",
    "import time\n",
    "import schedule\n",
    "def job_that_executes_once():\n",
    "    print('Hello')\n",
    "    return schedule.CancelJob\n",
    "schedule.every().minute.at(':34').do(job_that_executes_once)\n",
    "while True:\n",
    "    schedule.run_pending()\n",
    "    time.sleep(1)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 根据标签检索任务：\n",
    "\n",
    "# 检索所有任务：schedule.get_jobs()\n",
    "import schedule\n",
    "def greet(name):\n",
    "    print('Hello {}'.format(name))\n",
    "schedule.every().day.do(greet, 'Andrea').tag('daily-tasks', 'friend')\n",
    "schedule.every().hour.do(greet, 'John').tag('hourly-tasks', 'friend')\n",
    "schedule.every().hour.do(greet, 'Monica').tag('hourly-tasks', 'customer')\n",
    "schedule.every().day.do(greet, 'Derek').tag('daily-tasks', 'guest')\n",
    "friends = schedule.get_jobs('friend')\n",
    "print(friends)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 根据标签取消任务：\n",
    "\n",
    "# 取消所有任务：schedule.clear()\n",
    "import schedule\n",
    "def greet(name):\n",
    "    print('Hello {}'.format(name))\n",
    "    if name == 'Cancel':\n",
    "        schedule.clear('second-tasks')\n",
    "        print('cancel second-tasks')\n",
    "schedule.every().second.do(greet, 'Andrea').tag('second-tasks', 'friend')\n",
    "schedule.every().second.do(greet, 'John').tag('second-tasks', 'friend')\n",
    "schedule.every().hour.do(greet, 'Monica').tag('hourly-tasks', 'customer')\n",
    "schedule.every(5).seconds.do(greet, 'Cancel').tag('daily-tasks', 'guest')\n",
    "while True:\n",
    "    schedule.run_pending()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 运行任务到某时间：\n",
    "\n",
    "import schedule\n",
    "from datetime import datetime, timedelta, time\n",
    "def job():\n",
    "    print('working...')\n",
    "schedule.every().second.until('23:59').do(job)  # 今天23:59停止\n",
    "schedule.every().second.until('2030-01-01 18:30').do(job)  # 2030-01-01 18:30停止\n",
    "schedule.every().second.until(timedelta(hours=8)).do(job)  # 8小时后停止\n",
    "schedule.every().second.until(time(23, 59, 59)).do(job)  # 今天23:59:59停止\n",
    "schedule.every().second.until(datetime(2030, 1, 1, 18, 30, 0)).do(job)  # 2030-01-01 18:30停止\n",
    "while True:\n",
    "    schedule.run_pending()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 马上运行所有任务（主要用于测试）：\n",
    "\n",
    "import schedule\n",
    "def job():\n",
    "    print('working...')\n",
    "def job1():\n",
    "    print('Hello...')\n",
    "schedule.every().monday.at('12:40').do(job)\n",
    "schedule.every().tuesday.at('16:40').do(job1)\n",
    "schedule.run_all()\n",
    "schedule.run_all(delay_seconds=3)  # 任务间延迟3秒"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 并行运行：使用 Python 内置队列实现：\n",
    "\n",
    "import threading\n",
    "import time\n",
    "import schedule\n",
    "def job1():\n",
    "    print(\"I'm running on thread %s\" % threading.current_thread())\n",
    "def job2():\n",
    "    print(\"I'm running on thread %s\" % threading.current_thread())\n",
    "def job3():\n",
    "    print(\"I'm running on thread %s\" % threading.current_thread())\n",
    "def run_threaded(job_func):\n",
    "    job_thread = threading.Thread(target=job_func)\n",
    "    job_thread.start()\n",
    "schedule.every(10).seconds.do(run_threaded, job1)\n",
    "schedule.every(10).seconds.do(run_threaded, job2)\n",
    "schedule.every(10).seconds.do(run_threaded, job3)\n",
    "while True:\n",
    "    schedule.run_pending()\n",
    "    time.sleep(1)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 6. 利用任务框架APScheduler实现定时任务\n",
    "APScheduler（advanceded python scheduler）基于Quartz的一个Python定时任务框架，实现了Quartz的所有功能，使用起来十分方便。提供了基于日期、固定时间间隔以及crontab类型的任务，并且可以持久化任务。基于这些功能，我们可以很方便的实现一个Python定时任务系统。\n",
    "\n",
    "搜索公众号Linux中文社区猿后台回复“私房菜”，获取一份惊喜礼包。\n",
    "\n",
    "它有以下三个特点：\n",
    "\n",
    "类似于 Liunx Cron 的调度程序(可选的开始/结束时间)\n",
    "\n",
    "基于时间间隔的执行调度(周期性调度，可选的开始/结束时间)\n",
    "\n",
    "一次性执行任务(在设定的日期/时间运行一次任务)\n",
    "\n",
    "APScheduler有四种组成部分：\n",
    "\n",
    "触发器(trigger) 包含调度逻辑，每一个作业有它自己的触发器，用于决定接下来哪一个作业会运行。除了他们自己初始配置意外，触发器完全是无状态的。\n",
    "\n",
    "作业存储(job store) 存储被调度的作业，默认的作业存储是简单地把作业保存在内存中，其他的作业存储是将作业保存在数据库中。一个作业的数据讲在保存在持久化作业存储时被序列化，并在加载时被反序列化。调度器不能分享同一个作业存储。\n",
    "\n",
    "执行器(executor) 处理作业的运行，他们通常通过在作业中提交制定的可调用对象到一个线程或者进城池来进行。当作业完成时，执行器将会通知调度器。\n",
    "\n",
    "调度器(scheduler) 是其他的组成部分。你通常在应用只有一个调度器，应用的开发者通常不会直接处理作业存储、调度器和触发器，相反，调度器提供了处理这些的合适的接口。配置作业存储和执行器可以在调度器中完成，例如添加、修改和移除作业。通过配置executor、jobstore、trigger，使用线程池(ThreadPoolExecutor默认值20)或进程池(ProcessPoolExecutor 默认值5)并且默认最多3个(max_instances)任务实例同时运行，实现对job的增删改查等调度控制\n",
    "\n",
    "[image]\n",
    "\n",
    "示例代码："
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 7. 使用分布式消息系统Celery实现定时任务\n",
    "Celery是一个简单，灵活，可靠的分布式系统，用于处理大量消息，同时为操作提供维护此类系统所需的工具, 也可用于任务调度。Celery 的配置比较麻烦，如果你只是需要一个轻量级的调度工具，Celery 不会是一个好选择。\n",
    "\n",
    "Celery 是一个强大的分布式任务队列，它可以让任务的执行完全脱离主程序，甚至可以被分配到其他主机上运行。我们通常使用它来实现异步任务（async task）和定时任务（crontab）。异步任务比如是发送邮件、或者文件上传, 图像处理等等一些比较耗时的操作 ，定时任务是需要在特定时间执行的任务。\n",
    "\n",
    "需要注意，celery本身并不具备任务的存储功能，在调度任务的时候肯定是要把任务存起来的，因此在使用celery的时候还需要搭配一些具备存储、访问功能的工具，比如：消息队列、Redis缓存、数据库等。官方推荐的是消息队列RabbitMQ，有些时候使用Redis也是不错的选择。\n",
    "\n",
    "它的架构组成如下图："
   ],
   "metadata": {}
  }
 ],
 "metadata": {
  "orig_nbformat": 4,
  "language_info": {
   "name": "python",
   "version": "3.9.0",
   "mimetype": "text/x-python",
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "pygments_lexer": "ipython3",
   "nbconvert_exporter": "python",
   "file_extension": ".py"
  },
  "kernelspec": {
   "name": "python3",
   "display_name": "Python 3.9.0 64-bit ('universe': virtualenv)"
  },
  "interpreter": {
   "hash": "efc9d12c894566d142e82a9dae601d559d62f10cdf97a3f3ead6efac69d368e5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}